package com.liuyi.netty.im.server.handle.message;

import cn.hutool.core.util.StrUtil;
import com.liuyi.netty.im.commons.codes.CodeConst;
import com.liuyi.netty.im.commons.codes.LimProtocol;
import com.liuyi.netty.im.commons.codes.data.DataEvent;
import com.liuyi.netty.im.commons.codes.data.event.message.playload.Playload;
import com.liuyi.netty.im.server.handle.service.MessageListener;
import com.liuyi.netty.im.server.util.ImHelp;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.experimental.Accessors;
import lombok.extern.slf4j.Slf4j;

import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.locks.LockSupport;

/**
 * 消息发送处理器
 */
@Slf4j
public class SendMessageProcessor {

    /**
     * 创建一个消息队列
     */
    private final Queue<Message> messageQueue = new ConcurrentLinkedQueue<>();
    private final Thread thread;
    private final MessageListener messageListener;

    public SendMessageProcessor(MessageListener messageListener) {
        this.messageListener = messageListener;
        this.thread = new Thread(exec());
        this.thread.start();
    }

    private Runnable exec() {
        return new Runnable() {
            @Override
            public void run() {
                while (true) {
                    Message message = messageQueue.poll();
                    if (message != null) {
                        DataEvent dataEvent = message.getDataEvent();
                        LimProtocol protocol = ImHelp.getJsonProtocol(CodeConst.Command.EVENT, dataEvent);
                        Channel channel = message.getChannel();
                        if (channel != null && channel.isOpen()) {
                            ChannelFuture channelFuture = channel.writeAndFlush(protocol);
                            if (listener(dataEvent)) {
                                channelFuture.addListener(new MessageChannelFutureListener((com.liuyi.netty.im.commons.codes.data.event.Message<? extends Playload>) dataEvent.getData(),
                                        messageListener));
                            }
                        }
                    } else {
                        LockSupport.park();
                    }
                }
            }
        };
    }

    private boolean listener(DataEvent dataEvent) {
        if (!dataEvent.getEvent().equals(CodeConst.Event.MESSAGE_RECEIVED)) {
            return false;
        }
        com.liuyi.netty.im.commons.codes.data.event.Message<?> data = (com.liuyi.netty.im.commons.codes.data.event.Message<?>) dataEvent.getData();
        if (StrUtil.isEmpty(data.getMessageId())) {
            return false;
        }
        return true;
    }

    public boolean offer(Message message) {
        if (message.getChannel() != null && message.getChannel().isOpen()) {
            if (messageQueue.offer(message)) {
                LockSupport.unpark(thread);
                return true;
            }
            return false;
        }
        return true;
    }

    class MessageChannelFutureListener implements ChannelFutureListener {

        private final com.liuyi.netty.im.commons.codes.data.event.Message<? extends Playload> message;
        private final MessageListener messageListener;

        public MessageChannelFutureListener(com.liuyi.netty.im.commons.codes.data.event.Message<? extends Playload> message, MessageListener messageListener) {
            this.message = message;
            this.messageListener = messageListener;
        }

        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            if (future.isSuccess()) {
                messageListener.sendSuccess(message);
            } else {
                messageListener.sendFail(message);
            }
        }
    }


    @Data
    @Accessors
    @AllArgsConstructor
    public static class Message {

        private Channel channel;

        private DataEvent dataEvent;
    }
}
